package com.server.push.mqtt.base;

import com.server.push.mqtt.base.annotation.*;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.stereotype.Component;

import java.lang.reflect.Method;
import java.util.*;

@Component
@Slf4j
public class MqttProcessor {
    @Autowired
    MqttUtilsBean mqttUtilsBean;
    @Autowired
    MqttPahoMessageDrivenChannelAdapter adapter;
    List<String> topics = new ArrayList<>();

    Map<String,Method> methodsList = new HashMap<>();

    @Bean
    public void subscribe() {
        for (Object o : mqttUtilsBean.getControllerBeans()) {
            Method[] methods = o.getClass().getDeclaredMethods();
            for (Method method : methods) {

                method.setAccessible(true);
                MqttSubscribe mqttSubscribe = AnnotationUtils.findAnnotation(method, MqttSubscribe.class);
                if (mqttSubscribe != null) {
                    String topic = mqttSubscribe.topic();
                    int qos = mqttSubscribe.qos();
                    topic = Topic.getSubscribe(topic);
                    log.info("Mqtt 添加订阅: "+topic +"  qos:"+ qos);
                    topics.add(topic);
                    adapter.addTopic(topic,qos);
                    methodsList.put(topic,method);
                }


            }
        }


    }

    @SneakyThrows
    public void message(Message<?> message){

        String data = message.getPayload().toString();
        MessageHeaders messageHeaders = message.getHeaders();
        UUID packetId = messageHeaders.getId();
        Integer qos = (Integer) messageHeaders.get(MqttHeaders.RECEIVED_QOS);
        String receivedTopic = (String) messageHeaders.get(MqttHeaders.RECEIVED_TOPIC);
        if (receivedTopic ==null) return;

        for (Object bean : mqttUtilsBean.getControllerBeans()) {
            Method[] methods = bean.getClass().getDeclaredMethods();
            for (Method method : methods) {
                MqttSubscribe mqttSubscribe = AnnotationUtils.findAnnotation(method, MqttSubscribe.class);
                if (mqttSubscribe != null) {

                    method.setAccessible(true);
                    String originalTopic = mqttSubscribe.topic();
                    Topic topic = new Topic(receivedTopic,originalTopic);

                    // 匹配topic规则
                    if (!MqttTopic.isMatched(topic.getSubscribe(), receivedTopic)) continue;
                    /* 调用主方法 */
                    method.invoke(bean, topic,data);

                }
            }
        }

    }



}
